package com.zenitera.bigdata.transform;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * connect
 * 在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息
 * 这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。
 * <p>
 * Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后
 * 只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立
 * <p>
 * DataStream[A], DataStream[B] -> ConnectedStreams[A,B]
 */
public class Flink10_TransForm_Connect {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Integer> intStream = env.fromElements(1, 3, 5, 7, 9);
        DataStreamSource<String> stringStream = env.fromElements("two", "four", "six", "eight", "ten");
        ConnectedStreams<Integer, String> connectedStreams = intStream.connect(stringStream);
        connectedStreams.getFirstInput().print("流a:");
        connectedStreams.getSecondInput().print("流b:");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
